package com.zhao.demo.bound.demo05_broadcast.b_broadcast

import java.util

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration

import scala.collection.mutable

/**
 * Description: Dataset进行Join操作使用broadcast来进行优化<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/11/26 13:39
 *
 * @author 柒柒
 * @version : 1.0
 */

object BroadcastDemo {
  def main(args: Array[String]): Unit = {
    //需求：有两个有界流，其中一个有界流中存储的是性别信息，另外一个有界流存储的是学生的信息，需要将学生的信息完整显示
    //如：有界流→ (1,'男'),(2,'女')
    // 无界流 →(101,"jackson",1,"上海"),(104,"jone",2,"天津"), (108,"leon",1,"重庆")

    println(s"线程名:${Thread.currentThread().getName}")

    //1.环境
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    //2.获得两个DataSet
    val ds1: DataSet[(Int, Char)] = env.fromElements((1, '男'), (2, '女'))

    val ds2: DataSet[(Int, String, Int, String)] = env.fromElements((101, "jackson", 1, "上海"), (104, "jone", 2, "天津"), (108, "leon", 1, "重庆"))

    //3.将ds1作为广播变量中 往每个taskManager进程的内存中进行处理,并输出结果
    ds2.map(new RichMapFunction[(Int,String,Int,String),(Int,String,Char,String)] {
      //RichMapFunction实例中的方法运行在TaskManager中的slot管理的线程上

      /**
       * 准备容器,用来存放从广播变量中获取的值
       */
      var container: mutable.Map[Int,Char] = _

      /**
       * 用来进行初始化操作的,针对一个DataSet,该方法只会执行一次
       * @param parameters
       */
      override def open(parameters: Configuration): Unit = {

        //初始化全局变量container
        container = mutable.Map()

        //a)获取广播变量中封装的性别信息
        val lst: java.util.List[(Int,Char)] = getRuntimeContext.getBroadcastVariable("genderInformations")

        //b)将信息拿出来,存入到Map集合
        import scala.collection.JavaConversions._
        for (perEle <- lst){
          container.put(perEle._1,perEle._2)
        }
      }

      /**
       * 每次分析DataSet中的一个元素时,下述方法就会触发执行一次
       * @param value
       * @return
       */
      override def map(value: (Int, String, Int, String)): (Int, String, Char, String) = {
        println(s"TaskManger中的slot所管理的线程,线程名:${Thread.currentThread().getName}")
        val genderFlg = value._3
        val gender: Char = container.getOrElse(genderFlg, 'x')

        (value._1,value._2,gender,value._4)
      }
    }).withBroadcastSet(ds1,"genderInformations") //将ds1当成广播变量,分发到各个TaskManager的内存中
      .print()
  }
}
















